import org.apache.spark.sql.Row

import scala.collection.mutable.{Map => MutableMap}

/**
 * Utils静态方法
 */
object Utils {
  def transformOperatorFeatures(opList: Seq[String],
   operatorFeatures: Seq[OperatorFeatures]): Seq[Map[Long, Row]] = {
    val opDict = buildOperatorDict(opList, operatorFeatures)
    // 根据opList动态添加operator列
    opList.map { op =>
      val opKey = op + "_fm"
      opDict.get(opKey) match {
        case Some(fm) => fm.map { case (id, f) => id -> Row(f.value, f.updated_time) }
        case None => null
      }
    }  
  }

  def buildOperatorDict(opList: Seq[String], operatorFeatures: Seq[OperatorFeatures]): MutableMap[String, Map[Long, FeatureInfo]] = {
    val opDict = MutableMap[String, Map[Long, FeatureInfo]]()
    val opSet = opList.toSet
    operatorFeatures.foreach { opFeatures =>
      val opName = opFeatures.op_name.toLowerCase
      if (opSet.contains(opName)) {
        val features = opFeatures.features
        
        val fm = features.map { f =>
          // 将FeatureInfo转换为Row
          val featureRow = FeatureInfo(f.value, opFeatures.updated_time)
          f.id -> featureRow
        }.toMap
        val fmExist = opDict.get(opName + "_fm")
        if (fmExist.isDefined) {
          opDict(opName + "_fm") = Utils.mergeFeatureMapByLatestUpdateTime(fmExist.get, fm)
        } else {
          opDict(opName + "_fm") = fm
        }
      } else {
        // 使用println替代logger，避免序列化问题
        println(s"WARNING: new operators found: $opName, skip it")
      }
    } 
    opDict
  }
  def mergeFeatureMapByLatestUpdateTime(fm1: Map[Long, FeatureInfo], fm2: Map[Long, FeatureInfo]): Map[Long, FeatureInfo] = {
    val fmKeys = fm1.keySet ++ fm2.keySet
    fmKeys.map { key =>
      val f1 = fm1.get(key)
      val f2 = fm2.get(key)
      (key, if (f1.isDefined && f2.isDefined) {
        if (f1.get.updated_time > f2.get.updated_time) f1.get else f2.get
      } else if (f1.isDefined) {
        f1.get 
      } else {
        f2.get
      })
    }.toMap
  }
//   def updateAccountTimeHistoryRecord(row:Row): AccountTimeIndexRecord = {
//     val fetchTimeList = Set(row.getAs[Seq[Long]]("fetch_time_list"):_*).toSeq.sorted
//     val opHistory = row.getAs[Seq[Long]]("op_history")
//     val dayPart = row.getAs[Long]("day_part")
//     if (opHistory == null) {
//         AccountTimeIndexRecord(
//             account_id = row.getAs[Long]("account_id"),
//             op_name = row.getAs[String]("op_name"),
//             history_start = fetchTimeList.head,
//             history_end = fetchTimeList.last,
//             op_history = fetchTimeList,
//             day_part = dayPart
//         )
//     } else {
//         val record = AccountTimeIndexRecord(
//             account_id = row.getAs[Long]("account_id"),
//             op_name = row.getAs[String]("op_name"),
//             day_part = dayPart
//         )
//         val newOpHistory = Set(opHistory ++ fetchTimeList:_*).toSeq.sorted
//         record.history_start = newOpHistory.head
//         record.history_end = newOpHistory.last
//         record.op_history = newOpHistory
//         record
//     }
//   }
  /*  存在ABA问题，即如果A1, A2是2个时刻的结果，他们的value相等，此时A2不会插入。但是后面插入了一个B，value和A1不等，
      时刻在A1, A2中间，此时B会插入
     * 后面当我们按照时间搜索的时候，如果时刻正好在A2之后，那么返回的就变成了B，而不是A2
     * 给定2个输入
     * row1: Map{feature_id -> FeatureInfo(feature_value, updated_time)}
     * row2: Map{feature_id -> List[FeatureCell(feature_value, fetch_tiem, updated_time)]}, 其中row2
     *       中的featureCell列表是按照fetch_tiem排序的
     * 
     * 输出：
     * Map{feature_id -> List[FeatureCell(feature_value, fetch_time, updated_time)]}, 
     * 其中FeatureCell List中的某个Cell，是根据row1中对应的feature_id的value和time信息来更新的。
     * 具体逻辑如下
     *    1. 如果row1中feature_id在row2中不存在，则直接插入到Map中
     *    2. 如果feature id存在，则分2种情况
     *       2.1 如果row1和row2中feature_id的fetch_time相同，则根据updated_time选时间最新的value覆盖原value和updated_time
     *       2.2 否则，则根据fetch_time排序，先找到需要插入的位置，然后和前面一个位置的数据对比：
     *           2.2.1 如果前面一个位置的value和row1中的值相同，则不做任何处理
     *           2.2.2 否则，则插入到前面一个位置的后面 
     * 注意，返回的仅仅是通过featureMap更新后的key-value，并没有返回originFeatureMap所有的kv
     */
    // def updateAccountTimeOpHist(fetchTime: Long, 
    //         featureMap: Map[Long, FeatureInfo], 
    //         originFeatureMap: Map[Long, Seq[FeatureHistoryCell]]): Map[Long, Seq[FeatureHistoryCell]] = {
    //     featureMap.map { case (featureId, featureInfo) =>
    //         originFeatureMap.get(featureId) match {
    //             case Some(originFeatureCellList) => {
    //                 // 二分法查找在originFeatureCellList中，fetch_time小于等于fetchTime的最后一个位置
    //                 val insertIndex = binarySearchByFetchTime(originFeatureCellList, fetchTime)
    //                 if (insertIndex >= 0 && insertIndex < originFeatureCellList.length) {
    //                     // 情况2.1：如果fm1和fm2中feature_id的fetch_time相同，则根据updated_time选时间最新的value覆盖原value和updated_time
    //                     val existingCell = originFeatureCellList(insertIndex)
    //                     if (existingCell.fetch_time == fetchTime) { // 这里包含了insertPos == 0的情况
    //                         // fetch_time相同，比较updated_time
    //                         if (featureInfo.updated_time > existingCell.updated_time) {
    //                             // 新数据的updated_time更新，替换现有数据
    //                             val updatedList = originFeatureCellList.updated(insertIndex, 
    //                                 FeatureHistoryCell(featureInfo.value, fetchTime, featureInfo.updated_time))
    //                             (featureId, updatedList)
    //                         } else {
    //                             // 现有数据的updated_time更新，保持不变
    //                             (featureId, originFeatureCellList)
    //                         }
    //                     } else {
    //                         // 情况2.2：fetch_time不同，需要插入到合适位置
    //                         val prevPos = insertIndex-1
    //                         // 检查前面一个位置的数据
    //                         val prevCell = originFeatureCellList(prevPos)
    //                         if (prevCell.value == featureInfo.value) {
    //                             // 2.2.1 如果前面一个位置的value和row1中的值相同，则不做任何处理
    //                             (featureId, originFeatureCellList)
    //                         } else {
    //                             // 2.2.2 否则，则插入到前面一个位置的后面
    //                             val newCell = FeatureHistoryCell(featureInfo.value, fetchTime, featureInfo.updated_time)
    //                             val newList = originFeatureCellList.take(insertIndex) ++ Seq(newCell) ++ originFeatureCellList.drop(insertIndex)
    //                             (featureId, newList)
    //                         }
    //                     }
    //                 } else {
    //                     val newCell = FeatureHistoryCell(featureInfo.value, fetchTime, featureInfo.updated_time)
    //                     if (insertIndex < 0) {
    //                         // insertIndex < 0，说明targetFetchTime < history(left)，插入到开头
    //                         val nextCell = originFeatureCellList(0)
    //                         if (nextCell.value == featureInfo.value) {
    //                             // 2.2.1 如果第一个位置的value和之前head中的值相同，则替换原来的head
    //                             (featureId, Seq(newCell) ++ originFeatureCellList.drop(1) )
    //                         } else {
    //                             (featureId, Seq(newCell) ++ originFeatureCellList)
    //                         }
    //                     } else {
    //                         // insertIndex >= 0，说明targetFetchTime >= history(left)，插入到末尾
    //                         val prevCell = originFeatureCellList(originFeatureCellList.length - 1)
    //                         if (prevCell.value == featureInfo.value) {
    //                             // 2.2.1 如果前面一个位置的value和row1中的值相同，则不做任何处理
    //                             val lastCell = FeatureHistoryCell(featureInfo.value, 
    //                                             prevCell.fetch_time, 
    //                                             featureInfo.updated_time) // 这里更新updated_time，但保持fetch_time和原来一样
    //                             (featureId, originFeatureCellList.take(originFeatureCellList.length - 1)
    //                                              ++ Seq(lastCell))
    //                         } else {
    //                             (featureId, originFeatureCellList ++ Seq(newCell))
    //                         }
    //                     }
    //                 }
    //             }
    //             case None => {
    //                 // 情况1：如果row1中feature_id在row2中不存在，则直接插入到Map中
    //                 val newCell = FeatureHistoryCell(featureInfo.value, fetchTime, featureInfo.updated_time)
    //                 (featureId, Seq(newCell))
    //             }
    //         }
    //     }.toMap
    // }

}